package org.apache.rocketmq.store.index;

import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.config.StorePathConfigHelper;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/7/14
 * @方法描述：索引数据服务组件
 */
public class IndexService {

    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    //创建索引文件的最大尝试次数
    private static final int MAX_TRY_IDX_CREATE = 3;

    //默认消息存储引擎
    private final DefaultMessageStore defaultMessageStore;

    //索引文件可以使用的最大哈希槽数量
    private final int hashSlotNum;

    //索引文件可以存储的最大条目数
    private final int indexNum;

    //索引文件存储路径
    private final String storePath;

    //索引文件集合
    private final ArrayList<IndexFile> indexFileList = new ArrayList<>();

    private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    //构造方法
    public IndexService(final DefaultMessageStore store) {
        this.defaultMessageStore = store;
        this.hashSlotNum = store.getMessageStoreConfig().getMaxHashSlotNum();
        this.indexNum = store.getMessageStoreConfig().getMaxIndexNum();
        this.storePath = StorePathConfigHelper.getStorePathIndex(defaultMessageStore.getMessageStoreConfig().getStorePathRootDir());
    }
    //以下的部分方法我就不添加注释了，因为在之前的消费队列和CommitLog中都见过很多次了，逻辑几乎相同，大家自己看看就行
    //我只给索引文件独有的核心方法添加注释

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：加载索引文件的方法
     */
    public boolean load(final boolean lastExitOK) {
        File dir = new File(this.storePath);
        File[] files = dir.listFiles();
        if (files != null) {
            Arrays.sort(files);
            for (File file : files) {
                try {
                    IndexFile f = new IndexFile(file.getPath(), this.hashSlotNum, this.indexNum, 0, 0);
                    f.load();
                    if (!lastExitOK) {
                        if (f.getEndTimestamp() > this.defaultMessageStore.getStoreCheckpoint().getIndexMsgTimestamp()) {
                            f.destroy(0);
                            continue;
                        }
                    }
                    LOGGER.info("load index file OK, " + f.getFileName());
                    this.indexFileList.add(f);
                } catch (IOException e) {
                    LOGGER.error("load file {} error", file, e);
                    return false;
                } catch (NumberFormatException e) {
                    LOGGER.error("load file {} error", file, e);
                }
            }
        }
        return true;
    }

    //得到索引文件存储的总数据的方法
    public long getTotalSize() {
        if (indexFileList.isEmpty()) {
            return 0;
        }
        return (long) indexFileList.get(0).getFileSize() * indexFileList.size();
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：删除过期的索引文件的方法
     */
    public void deleteExpiredFile(long offset) {
        Object[] files = null;
        try {
            this.readWriteLock.readLock().lock();
            if (this.indexFileList.isEmpty()) {
                return;
            }

            long endPhyOffset = this.indexFileList.get(0).getEndPhyOffset();
            if (endPhyOffset < offset) {
                files = this.indexFileList.toArray();
            }
        } catch (Exception e) {
            LOGGER.error("destroy exception", e);
        } finally {
            this.readWriteLock.readLock().unlock();
        }

        if (files != null) {
            List<IndexFile> fileList = new ArrayList<>();
            for (int i = 0; i < (files.length - 1); i++) {
                IndexFile f = (IndexFile) files[i];
                if (f.getEndPhyOffset() < offset) {
                    fileList.add(f);
                } else {
                    break;
                }
            }

            this.deleteExpiredFile(fileList);
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：删除过期的索引文件的方法
     */
    private void deleteExpiredFile(List<IndexFile> files) {
        if (!files.isEmpty()) {
            try {
                this.readWriteLock.writeLock().lock();
                for (IndexFile file : files) {
                    boolean destroyed = file.destroy(3000);
                    destroyed = destroyed && this.indexFileList.remove(file);
                    if (!destroyed) {
                        LOGGER.error("deleteExpiredFile remove failed.");
                        break;
                    }
                }
            } catch (Exception e) {
                LOGGER.error("deleteExpiredFile has exception.", e);
            } finally {
                this.readWriteLock.writeLock().unlock();
            }
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消费索引文件的方法
     */
    public void destroy() {
        try {
            this.readWriteLock.writeLock().lock();
            for (IndexFile f : this.indexFileList) {
                f.destroy(1000 * 3);
            }
            this.indexFileList.clear();
        } catch (Exception e) {
            LOGGER.error("destroy exception", e);
        } finally {
            this.readWriteLock.writeLock().unlock();
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：查询索引文件中指定主题，key和时间范围内存消息的方法
     */
    public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long begin, long end) {
        List<Long> phyOffsets = new ArrayList<>(maxNum);
        long indexLastUpdateTimestamp = 0;
        long indexLastUpdatePhyoffset = 0;
        maxNum = Math.min(maxNum, this.defaultMessageStore.getMessageStoreConfig().getMaxMsgsNumBatch());
        try {
            this.readWriteLock.readLock().lock();
            if (!this.indexFileList.isEmpty()) {
                for (int i = this.indexFileList.size(); i > 0; i--) {
                    IndexFile f = this.indexFileList.get(i - 1);
                    boolean lastFile = i == this.indexFileList.size();
                    if (lastFile) {
                        indexLastUpdateTimestamp = f.getEndTimestamp();
                        indexLastUpdatePhyoffset = f.getEndPhyOffset();
                    }
                    if (f.isTimeMatched(begin, end)) {

                        f.selectPhyOffset(phyOffsets, buildKey(topic, key), maxNum, begin, end);
                    }
                    if (f.getBeginTimestamp() < begin) {
                        break;
                    }
                    if (phyOffsets.size() >= maxNum) {
                        break;
                    }
                }
            }
        } catch (Exception e) {
            LOGGER.error("queryMsg exception", e);
        } finally {
            this.readWriteLock.readLock().unlock();
        }
        return new QueryOffsetResult(phyOffsets, indexLastUpdateTimestamp, indexLastUpdatePhyoffset);
    }



    //构建索引key的方法
    private String buildKey(final String topic, final String key) {
        return topic + "#" + key;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把消息存放到索引文件中的方法
     */
    public void buildIndex(DispatchRequest req) {
        //尝试得到一个可用的索引文件
        IndexFile indexFile = retryGetAndCreateIndexFile();
        if (indexFile != null) {
            //成功获取到了索引文件之后，得到该索引文件存储的最后一条消息的物理偏移量，也就是这条消息在CommitLog文件中的全局偏移量
            long endPhyOffset = indexFile.getEndPhyOffset();
            DispatchRequest msg = req;
            //得到要存储的消息的主题
            String topic = msg.getTopic();
            //得到消息的key集合
            String keys = msg.getKeys();
            //判断当前要存储的消息在CommitLog文件中的全局偏移量是否小于索引文件的最新消息全局偏移量
            //如果小于则意味着这条消息过期了，不用存储，直接退出当前方法即可
            if (msg.getCommitLogOffset() < endPhyOffset) {
                return;
            }
            //得到消息的事务类型
            final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
            //根据事务类型决定是否构建索引
            switch (tranType) {
                case MessageSysFlag.TRANSACTION_NOT_TYPE:
                case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                    break;
                //以上三种类型消息都会正常存储消息
                //事务回滚消息则不存储，直接退出当前方法
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                    return;
            }
            //判断消息的唯一键是否不为空
            if (req.getUniqKey() != null) {
                //不为空则使用消息唯一键和主题构建key，然后把消息存储索引文件
                indexFile = putKey(indexFile, msg, buildKey(topic, req.getUniqKey()));
                if (indexFile == null) {
                    //存储失败则记录错误日志
                    LOGGER.error("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset(), req.getUniqKey());
                    return;
                }
            }
            //判断消息是否拥有key
            if (keys != null && keys.length() > 0) {
                //如果有则分割key字符串，得到key数组
                String[] keyset = keys.split(MessageConst.KEY_SEPARATOR);
                //遍历所有key
                for (int i = 0; i < keyset.length; i++) {
                    //得到当前便利的key
                    String key = keyset[i];
                    //判断key是否有效
                    if (key.length() > 0) {
                        //使用主题和key构建新的key，然后把消息存储到索引文件中
                        indexFile = putKey(indexFile, msg, buildKey(topic, key));
                        if (indexFile == null) {
                            //存储失败则记录错误日志
                            LOGGER.error("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset(), req.getUniqKey());
                            return;
                        }
                    }
                }
            }
        } else {//没有获取可以存储消息的索引文件则记录错误日志
            LOGGER.error("build index error, stop building index");
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把消息存放到索引文件中的方法
     */
    private IndexFile putKey(IndexFile indexFile, DispatchRequest msg, String idxKey) {
        //在这里把消息放到索引文件中
        for (boolean ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp()); !ok; ) {
            //存放失败则记录文件已满的日志信息
            LOGGER.warn("Index file [" + indexFile.getFileName() + "] is full, trying to create another one");
            //尝试获取或创建新的索引文件
            indexFile = retryGetAndCreateIndexFile();
            //新的索引文件创建失败则直接返回null
            if (null == indexFile) {
                return null;
            }
            //再次存储消息到索引文件中，再次失败会再次循环
            ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp());
        }
        //存储成功则返回索引文件本身
        return indexFile;
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：重新获取或者创建新的索引文件的方法
     */
    public IndexFile retryGetAndCreateIndexFile() {
        IndexFile indexFile = null;
        //在循环中创建索引文件，循环次数为最大重试次数
        for (int times = 0; null == indexFile && times < MAX_TRY_IDX_CREATE; times++) {
            //获取最新索引文件，获取失败则创建新的索引文件
            indexFile = this.getAndCreateLastIndexFile();
            //获取成功直接退出循环
            if (null != indexFile) {
                break;
            }
            try {
                //获取失败则记录尝试创建索引文件次数的日志信息
                LOGGER.info("Tried to create index file " + times + " times");
                //等待1秒后再次尝试
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                LOGGER.error("Interrupted", e);
            }
        }
        //循环结束仍没有成功创建索引文件
        if (null == indexFile) {
            //那就给消息存储引擎设置索引文件状态错误的标志
            this.defaultMessageStore.getRunningFlags().makeIndexFileError();
            LOGGER.error("Mark index file cannot build flag");
        }
        return indexFile;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：获取最新索引文件，获取失败则创建新的索引文件的方法
     */
    public IndexFile getAndCreateLastIndexFile() {
        //记录当前得到的索引文件的变量
        IndexFile indexFile = null;
        //记录上一个索引文件的变量
        IndexFile prevIndexFile = null;
        //记录上一个索引文件的结束物理偏移量
        long lastUpdateEndPhyOffset = 0;
        //记录上一个索引文件存储最后消息的时间戳
        long lastUpdateIndexTimestamp = 0;
        {
            //获取读锁
            this.readWriteLock.readLock().lock();
            //判断索引文件集合是否不为空
            if (!this.indexFileList.isEmpty()) {
                //如果不为空则先获取最后一个索引文件
                IndexFile tmp = this.indexFileList.get(this.indexFileList.size() - 1);
                //判断最后一个索引文件是否写满了
                if (!tmp.isWriteFull()) {
                    //没有写满则可以直接返回
                    indexFile = tmp;
                } else {
                    //如果写满了，则要创建新的索引文件，这里得到旧的索引文件的的结束物理偏移量和存储最后消息的时间戳
                    lastUpdateEndPhyOffset = tmp.getEndPhyOffset();
                    lastUpdateIndexTimestamp = tmp.getEndTimestamp();
                    //记录旧的索引问文件
                    prevIndexFile = tmp;
                }
            }
            //释放读锁
            this.readWriteLock.readLock().unlock();
        }
        if (indexFile == null) {
            try {
                //得到要创建的新的索引文件名称
                String fileName = this.storePath + File.separator + UtilAll.timeMillisToHumanString(System.currentTimeMillis());
                //创建新的索引文件，索引文件的起始物理偏移量和起始时间戳要根据上一个文件的结束物理偏移量和存储最后消息的时间戳来初始化
                //保证索引文件存储数据的连续性
                indexFile = new IndexFile(fileName, this.hashSlotNum, this.indexNum, lastUpdateEndPhyOffset, lastUpdateIndexTimestamp);
                //得到写锁
                this.readWriteLock.writeLock().lock();
                //把新创建的索引文件加入到集合中
                this.indexFileList.add(indexFile);
            } catch (Exception e) {
                LOGGER.error("getLastIndexFile exception ", e);
            } finally {
                //释放写锁
                this.readWriteLock.writeLock().unlock();
            }
            if (indexFile != null) {
                //索引文件创建成功了，则需要把上一个索引文件的数据刷新到硬盘
                //这里得到上一个索引文件
                final IndexFile flushThisFile = prevIndexFile;
                //创建新的线程
                Thread flushThread = new Thread(new AbstractBrokerRunnable(defaultMessageStore.getBrokerConfig()) {
                    @Override
                    public void run0() {
                        //在新线程中执行索引文件刷新数据的操作
                        IndexService.this.flush(flushThisFile);
                    }
                }, "FlushIndexFileThread");
                //设置线程为守护线程
                flushThread.setDaemon(true);
                //启动刷盘线程
                flushThread.start();
            }
        }
        //返回创建的新文件，创建失败则返回null
        return indexFile;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把索引文件数据刷新到硬盘的方法
     */
    public void flush(final IndexFile f) {
        if (null == f) {
            return;
        }
        long indexMsgTimestamp = 0;
        if (f.isWriteFull()) {
            indexMsgTimestamp = f.getEndTimestamp();
        }
        f.flush();
        if (indexMsgTimestamp > 0) {
            this.defaultMessageStore.getStoreCheckpoint().setIndexMsgTimestamp(indexMsgTimestamp);
            this.defaultMessageStore.getStoreCheckpoint().flush();
        }
    }


    //启动索引文件服务的方法
    public void start() {
        //源码中该方法就是空方法
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：关闭索引服务的方法
     */
    public void shutdown() {
        try {
            this.readWriteLock.writeLock().lock();
            for (IndexFile f : this.indexFileList) {
                try {
                    f.shutdown();
                } catch (Exception e) {
                    LOGGER.error("shutdown " + f.getFileName() + " exception", e);
                }
            }
            this.indexFileList.clear();
        } catch (Exception e) {
            LOGGER.error("shutdown exception", e);
        } finally {
            this.readWriteLock.writeLock().unlock();
        }
    }
}